package election.platform;

import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Date;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import election.api.Get;
import election.api.OpMap;
import election.api.OpReduce;
import election.api.OpSpawn;



public class Platform
{
    static LocalService    locals;
    static OpSpace         ops;
    static DataSpace       datas;
    
    static ExecutorService    pool = Executors.newFixedThreadPool(10);
    
    public static void main(String[] args) throws Exception
    {
        locals = new LocalService(9090);
        ops = new OpSpace(9080);
        datas = new DataSpace(9070);
        
        
        pool.submit(locals);
        pool.submit(ops);
        pool.submit(datas);

        while (true) {
            Thread.sleep(10000);
            System.out.println(new Date());
            System.out.println("Data: " + datas.datasets.entrySet());
            System.out.println("Spawn: " + ops.nullOps);
            System.out.println("Reduce: " + ops.ops);
            
            // 1. first do nullOps (operation that does not need input)
            while (ops.nullOps.size() > 0) {
                Operation op = ops.nullOps.remove(0);
                OpSpawn spawn = (OpSpawn)op.operator;
                String outputSet = op.output;
                String nextHost = op.nextHost;
                doNullOp(spawn, outputSet, nextHost);
                
            }
            
            
            Iterator<Entry<String,MyQueue>> itr = datas.datasets.entrySet().iterator();
            
            while (itr.hasNext()) try {
                // 2. work with operation that need input
                
                Entry<String,MyQueue> entry = itr.next();
                String dataset = entry.getKey();
                Operation op = ops.take(dataset);
                if (op == null) {
                    System.err.println("[Dataset: "+dataset+"] no op exists");
                    continue;
                }
                MyQueue queue = entry.getValue();

                assert op.operator instanceof OpReduce;
                OpReduce reduce = (OpReduce)op.operator;
                reduce.apply(new CountdownGet(queue,op.feedCount), null);      // assume there is Reduce has no exit value
                datas.clear(dataset);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        
    }


    private static void doNullOp(OpSpawn spawn, String outputSet, String nextHost) throws Exception
    {
        MyQueue queue = new MyQueue();
        spawn.apply(queue);
        if (outputSet != null && nextHost != null) {
            Socket conn = new Socket(nextHost,9070);
            ObjectOutputStream out = new ObjectOutputStream(conn.getOutputStream());
            out.writeUTF(outputSet);
            for (Object obj = queue.get(); obj != null; obj = queue.get())
                out.writeObject(obj);
            conn.close();
        }
    }
    
    
    
}
